package cn.tedu.batch.transformation

import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint
import org.apache.flink.api.scala.ExecutionEnvironment

/**
 * @author Amos
 * @date 2022/5/20
 */

object BatchJoinDemo {
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    import org.apache.flink.api.scala._

    val subjectSource: DataSet[Subject] = env.readCsvFile[Subject]("FLINKSCALA/data/subject.csv")

    val scoreSource: DataSet[Score] = env.readCsvFile[Score]("FLINKSCALA/data/score.csv")

    val result: JoinDataSet[Subject, Score] = subjectSource.join(scoreSource,JoinHint.BROADCAST_HASH_FIRST).where(_.subid).equalTo(_.subid)

    result.print()

  }

}

case class Subject(subid: Int, subName: String)

case class Score(id: Int, stuName: String, subid: Int, socre: Double)
